package com.qqt.csr.im.mq;

import com.alibaba.fastjson2.JSONObject;
import com.corundumstudio.socketio.SocketIOClient;
import com.qqt.csr.common.utils.JsonUtil;
import com.qqt.csr.im.entity.mq.MqMessage;
import com.qqt.csr.im.enums.MsgTypeEnum;
import com.qqt.csr.im.service.EventMsgProducerService;
import com.qqt.csr.im.service.mongo.MessageService;
import com.qqt.csr.im.socketio.constants.EventsConstants;
import com.qqt.csr.im.socketio.utils.SocketIOLocalCacheUtil;
import com.qqt.csr.im.utils.ServiceHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * @Author: yang hao
 * @Date: 2024/10/15 15:51
 * @Description: 发送消息事件监听器
 **/
@Slf4j
@Service
@RocketMQMessageListener(
        consumerGroup = "${rocketmq.consumer.group}" + "_" + EventsConstants.S2S_MESSAGE_EVENT,
        topic = EventMsgProducerService.EVENT_FORWARD_TOPIC,
        selectorType = SelectorType.TAG,
        selectorExpression = EventsConstants.S2S_MESSAGE_EVENT,
        messageModel = MessageModel.BROADCASTING,
        tlsEnable = "${rocketmq.consumer.tlsEnable}"
)
public class SendMessageListener implements RocketMQListener<MqMessage> {
    @Autowired
    private MessageService messageService;

    @Override
    public void onMessage(MqMessage data) {
        if (ServiceHelper.INSTANCE.getServiceNode() == null) {
            return;
        }
        // 过滤服务id
        if (!data.getServiceId().equals(ServiceHelper.INSTANCE.getServiceNode().getId())) {
            return;
        }

        SocketIOClient client = SocketIOLocalCacheUtil.getClient(data.getDeviceKey());
        if (client == null || !client.isChannelOpen()) {
            log.info("socket已断开连接，推送消息失败！key:{},msg:{}", data.getDeviceKey(), JsonUtil.toJson(data));
            return;
        }

        try {
            client.sendEvent(EventsConstants.S2C_MESSAGE_EVENT, data.getMessage());
            log.info("socket已推送消息，key:{},clientId:{},conversationId:{}", data.getDeviceKey(), client.getSessionId(), data.getMessage().getConversationId());
            if (!MsgTypeEnum.SystemElem.equals(data.getMessage().getType())) {
                // 非系统消息，推送未读通知
                long count = messageService.getNoReadCount(data.getMessage().getConversationId(), data.getMessage().getRecipient().getId());
                JSONObject content = new JSONObject();
                content.put("roomId",data.getMessage().getRoomId());
                content.put("conversationId",data.getMessage().getConversationId());
                content.put("noReadCount",count);
                client.sendEvent(EventsConstants.NO_READ_EVENT, content);
            }
        } catch (Exception e) {
            log.error("推送消息失败！data:{}", JsonUtil.toJson(data), e);
        }
    }

}
